package com.thread.laoma;

import java.util.ArrayDeque;
import java.util.Queue;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

/**
* 生产者消费者模式二（使用显示锁 ）
 * Created by zhuangqi on 17-4-11.
 */
public class PcSyncDemo2 {

    static class MyBlokingQueue<E>{
        private Queue<E> queue = null;
        private int limit;
        //显示锁
        private Lock lock = new ReentrantLock();
        //显示条件
        private Condition notFull = lock.newCondition();
        private Condition notEmpty = lock.newCondition();

        public MyBlokingQueue(int limit) {
            this.queue = new ArrayDeque<>(limit);
            this.limit = limit;
        }

        public void put(E e) throws InterruptedException {
            lock.lockInterruptibly();
            try {
                while (queue.size() == limit){
                    notFull.await();
                }
                queue.add(e);
                notEmpty.signal();
            }finally {
                lock.unlock();
            }
        }

        public E take() throws InterruptedException {
            lock.lockInterruptibly();
            try {
                while (queue.isEmpty()){
                    notEmpty.await();
                }
                E data = queue.poll();
                notEmpty.signal();
                return data;
            }finally {
                lock.unlock();
            }
        }
    }

    static class Product extends Thread{
        private MyBlokingQueue<String> myBlokingQueue;
        public Product(MyBlokingQueue<String> myBlokingQueue) {
            this.myBlokingQueue = myBlokingQueue;
        }
        @Override
        public void run() {
            int num = 0;
            while (true){
                try {
                    String task = String.valueOf(num);
                    myBlokingQueue.put(task);
                    num++;
                    System.out.println("生产："+num);
                    Thread.sleep((int) (Math.random() * 100));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    static class Consumer extends Thread{
        private MyBlokingQueue<String> myBlokingQueue;
        public Consumer(MyBlokingQueue<String> myBlokingQueue) {
            this.myBlokingQueue = myBlokingQueue;
        }
        @Override
        public void run() {
            while (true){
                try {
                    String data = myBlokingQueue.take();
                    System.out.println("消费："+ data);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) {
        int sum = 10;
        MyBlokingQueue<String> queue = new MyBlokingQueue<>(sum);
        Thread thread = new Product(queue);
        Thread thread1 = new Consumer(queue);
        thread.start();
        thread1.start();
    }
}
